package com.boarsoft.message.core;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.xml.ResourceEntityResolver;
import org.springframework.beans.factory.xml.XmlBeanDefinitionReader;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.core.io.support.EncodedResource;

import com.boarsoft.bean.ReplyInfo;
import com.boarsoft.common.Util;
import com.boarsoft.common.util.FileUtil;
import com.boarsoft.message.MessageConfig;
import com.boarsoft.message.bean.RecipientMQ;
import com.boarsoft.message.bean.TargetMQ;
import com.boarsoft.message.store.MessageRoller;

//@Component("messageContext")
public class MessageContext {
	private static final Logger log = LoggerFactory.getLogger(MessageContext.class);

	/** */
	@Autowired
	protected DefaultListableBeanFactory beanFactory;
	/** */
	@Autowired
	protected ConfigurableApplicationContext appCtx;

	protected ExecutorService threadPool = Executors.newCachedThreadPool();

	/** */
	protected String charset = "UTF-8";
	/** Spring中配置的target队列，k: Queue/Topic name，代表一个种类型的消息 */
	protected Map<String, TargetMQ> targetQueueMap = new ConcurrentHashMap<String, TargetMQ>();
	/** k: Recipient，对应一个 MessageService */
	protected Map<String, RecipientMQ> recipientQueueMap = new ConcurrentHashMap<String, RecipientMQ>();

	/** 用于保存被暂停的目标队列 */
	protected Map<String, TargetMQ> _targetQueueMap = new HashMap<String, TargetMQ>();
	/** 用于保存移除中的收件人队列 */
	protected Map<String, RecipientMQ> _recipientQueueMap = new HashMap<String, RecipientMQ>();

	/** */
	protected final Map<String, MessageService> serviceMap = new ConcurrentHashMap<String, MessageService>();

	@PostConstruct
	public void init() {
		// 初始化收件人队列，启动处理线程
		if (recipientQueueMap.isEmpty()) {
			recipientQueueMap.putAll(appCtx.getBeansOfType(RecipientMQ.class));
		}
		for (Entry<String, RecipientMQ> en : recipientQueueMap.entrySet()) {
			String key = en.getKey();
			RecipientMQ rq = en.getValue();
			rq.init();
			if (Util.strIsEmpty(rq.getCode())) {
				rq.setCode(key);
			}
			this.addRecipientQueue(key, rq);
		}
		// 遍历所有targetQueue，有roller就初始化之（检查相关的目录或表是否需创建）
		if (targetQueueMap.isEmpty()) {
			targetQueueMap.putAll(appCtx.getBeansOfType(TargetMQ.class));
		}
		for (Entry<String, TargetMQ> en : targetQueueMap.entrySet()) {
			String k = en.getKey();
			final TargetMQ tq = en.getValue();
			tq.init();
			if (Util.strIsEmpty(tq.getCode())) {
				tq.setCode(k); // 如果不指定code，则用map的key
			}
			this.activeTargetQueue(tq);
		}
	}

	protected ReplyInfo<Object> activeTargetQueue(TargetMQ tq) {
		MessageRoller mr = tq.getRoller();
		if (mr == null) {
			return ReplyInfo.SUCCESS;
		}
		try {
			mr.init();
		} catch (Exception e) {
			log.error("Error on init roller of MQ {}", tq, e);
			return new ReplyInfo<Object>(e.getMessage());
		}
		return ReplyInfo.SUCCESS;
	}

	public ReplyInfo<Object> activeTargetQueue(String key) {
		TargetMQ tq = _targetQueueMap.remove(key);
		if (tq == null) {
			tq = targetQueueMap.get(key);
			if (tq == null) {
				return new ReplyInfo<Object>(//
						String.format("Target queue %s does not exists", key));
			}
		} else {
			targetQueueMap.put(key, tq);
		}
		return this.activeTargetQueue(tq);
	}

	public ReplyInfo<Object> deactiveTargetQueue(String key) {
		TargetMQ tq = targetQueueMap.remove(key);
		if (tq == null) {
			return new ReplyInfo<Object>(//
					String.format("Target queue %s does not exists", key));
		}
		_targetQueueMap.put(key, tq);
		return ReplyInfo.SUCCESS;
	}

	protected ReplyInfo<Object> activeRecipientQueue(RecipientMQ rq) {
		MessageRoller mr = rq.getRoller();
		if (mr == null) {
			return ReplyInfo.SUCCESS;
		}
		try {
			mr.init();
		} catch (Exception e) {
			log.error("Error on init roller of recipient queue {}", rq, e);
			return new ReplyInfo<Object>(e.getMessage());
		}
		return ReplyInfo.SUCCESS;
	}

	public ReplyInfo<Object> activeRecipientQueue(String key) {
		RecipientMQ rq = _recipientQueueMap.remove(key);
		if (rq == null) {
			rq = recipientQueueMap.get(key);
			if (rq == null) {
				return new ReplyInfo<Object>(//
						String.format("Recipient queue %s does not exists", key));
			}
		} else {
			recipientQueueMap.put(key, rq);
		}
		return this.activeRecipientQueue(rq);
	}

	public ReplyInfo<Object> deactiveRecipientQueue(String key) {
		RecipientMQ rq = recipientQueueMap.remove(key);
		if (rq == null) {
			return new ReplyInfo<Object>(//
					String.format("Recipient queue %s does not exists", key));
		}
		_recipientQueueMap.put(key, rq);
		return ReplyInfo.SUCCESS;
	}

	public ReplyInfo<Object> removeTargetQueue(String key, boolean force) {
		TargetMQ tq = targetQueueMap.remove(key);
		if (tq == null) {
			return new ReplyInfo<Object>(//
					String.format("Target queue %s does not exists", key));
		}
		if (tq.isEmpty()) {
			return ReplyInfo.SUCCESS;
		}
		if (force) {
			log.warn("Discard {} messages from removed target queue {}", tq.size(), key);
			tq.getQueue().clear();
			return ReplyInfo.SUCCESS;
		}
		// 如果队列不为空，则作为“已删除队列”的临时队列继续存在，以便MessageCore清空它
		String k = Util.date2str("yyyyMMddhhMMssSSS");
		k = String.format("removed_%s_%s", key, k);
		targetQueueMap.put(k, tq);
		return ReplyInfo.SUCCESS;
	}

	public ReplyInfo<Object> removeRecipientQueue(String key, boolean force) {
		RecipientMQ rq = recipientQueueMap.remove(key);
		if (rq == null) {
			return new ReplyInfo<Object>(//
					String.format("Recipient queue %s does not exists", key));
		}
		if (force) {
			log.warn("Discard {} messages from removed recipient queue {}", rq.size(), key);
			rq.getQueue().clear();
		}
		if (rq.isEmpty()) {
			rq.stop();
			// 移除队列事件通知

			return ReplyInfo.SUCCESS;
		}
		// 将此收件人队列移到临时表，以便在其消息被清空后，再停止此线程
		_recipientQueueMap.put(key, rq);
		return ReplyInfo.SUCCESS;
	}

	protected ReplyInfo<Object> loadQueueConfig(String xml) throws UnsupportedEncodingException {
		if (Util.strIsEmpty(xml)) {
			return ReplyInfo.SUCCESS;
		}
		// 将XML转换为BEAN
		EncodedResource er = new EncodedResource(//
				new ByteArrayResource(xml.getBytes(charset)));
		XmlBeanDefinitionReader xbdr = new XmlBeanDefinitionReader(beanFactory);
		xbdr.setResourceLoader(appCtx);
		xbdr.setEntityResolver(new ResourceEntityResolver(appCtx));
		try {
			log.debug("Load queue config bean from xml:\n{}\n", xml);
			int beans = xbdr.loadBeanDefinitions(er);
			log.info("Load {} queue config bean from xml successfully", beans);
		} catch (Exception e) {
			log.error("Invalid bean definitions xml: {}", xml, e);
			return new ReplyInfo<Object>("Invalid bean definitions xml");
		}
		return ReplyInfo.SUCCESS;
	}

	public ReplyInfo<Object> addTargetQueue(String key, TargetMQ tq) {
		// 添加到targetQueueMap并激活
		tq.init();
		targetQueueMap.put(key, tq);
		MessageRoller mr = tq.getRoller();
		if (mr == null) {
			return ReplyInfo.SUCCESS;
		}
		try {
			mr.init();
		} catch (Exception e) {
			log.error("Error on init roller of target queue {}", tq, e);
			return new ReplyInfo<Object>(e.getMessage());
		}
		return ReplyInfo.SUCCESS;
	}

	public ReplyInfo<Object> addTargetQueue(String key, String xml) {
		if (targetQueueMap.containsKey(key)) {
			return new ReplyInfo<Object>(String.format(//
					"Target queue %s already exists", key));
		}
		String fp = new StringBuilder(MessageConfig.APP_ROOT)//
				.append(MessageConfig.CONF_DIR).append("/tq_")//
				.append(key).append(".xml").toString();
		try {
			// 先尝试装载，装载成功再文件
			this.loadQueueConfig(xml);
			// 将收到的XML字符串写入以key为名的配置文件
			FileUtil.write(new File(fp), xml);
		} catch (UnsupportedEncodingException e) {
			return new ReplyInfo<Object>(String.format(//
					"Target queue %s config xml require charset {}", key, charset));
		} catch (FileNotFoundException e) {
			return new ReplyInfo<Object>(String.format(//
					"File path '%s' is invalid", fp));
		}
		if (!appCtx.containsBean(key)) {
			return new ReplyInfo<Object>(String.format(//
					"Target queue %s does not exists in received xml", key));
		}
		TargetMQ tq = (TargetMQ) appCtx.getBean(key);
		return this.addTargetQueue(key, tq);
	}

	public ReplyInfo<Object> updateTargetQueue(String key, String xml) {
		this.removeTargetQueue(key, false);
		this.addTargetQueue(key, xml);
		return ReplyInfo.SUCCESS;
	}

	public ReplyInfo<Object> addRecipientQueue(String key, String xml) {
		if (recipientQueueMap.containsKey(key)) {
			return new ReplyInfo<Object>(//
					String.format("Recipient queue %s already exists", key));
		}
		String fp = new StringBuilder(MessageConfig.APP_ROOT)//
				.append(MessageConfig.CONF_DIR).append("/rq_")//
				.append(key).append(".xml").toString();
		try {
			this.loadQueueConfig(xml);
			// 将收到的XML字符串写入以key为名的配置文件
			FileUtil.write(new File(fp), xml);
		} catch (UnsupportedEncodingException e) {
			return new ReplyInfo<Object>(String.format(//
					"Recipient queue %s config xml require charset {}", key, charset));
		} catch (FileNotFoundException e) {
			return new ReplyInfo<Object>(String.format(//
					"File path '%s' is invalid", fp));
		}
		// 将收到的XML字符串写入以key为名的配置文件并装载
		if (!appCtx.containsBean(key)) {
			return new ReplyInfo<Object>(String.format(//
					"Recipient queue %s does not exists in received xml", key));
		}
		RecipientMQ rq = (RecipientMQ) appCtx.getBean(key);
		// 添加到recipientQueueMap并激活
		return this.addRecipientQueue(key, rq);
	}

	public ReplyInfo<Object> updateRecipientQueue(String key, String xml) {
		this.removeRecipientQueue(key, false);
		this.addRecipientQueue(key, xml);
		return ReplyInfo.SUCCESS;
	}

	/**
	 * 返回预置的target队列
	 * 
	 * @param target
	 * @return
	 */
	public TargetMQ getTargetQueue(String target) {
		TargetMQ tq = targetQueueMap.get(target);
		if (tq == null) {
			throw new IllegalStateException(String.format(//
					"Target queue %s does not exists", target));
		}
		return tq;
	}

	/**
	 * 清除被移除的空目标队列和空收件人队列。
	 */
	public void clear() {
		for (Entry<String, TargetMQ> en : _targetQueueMap.entrySet()) {
			String k = en.getKey();
			TargetMQ tq = en.getValue();
			if (tq.isEmpty()) {
				_targetQueueMap.remove(k);
			}
		}
		for (Entry<String, RecipientMQ> en : _recipientQueueMap.entrySet()) {
			String k = en.getKey();
			RecipientMQ rq = en.getValue();
			if (rq.isEmpty()) {
				rq.stop(); // 退出线程
				_recipientQueueMap.remove(k);
			}
		}
	}

	public RecipientMQ getRecipientQueue(String recipient) {
		return recipientQueueMap.get(recipient);
	}

	public ReplyInfo<Object> addRecipientQueue(String key, RecipientMQ rq) {
		// 动态装载的recipientMQ可能没有被注入messageContext
		rq.setContext(this);
		rq.init();
		recipientQueueMap.put(key, rq);
		//
		MessageRoller mr = rq.getRoller();
		if (mr != null) {
			try {
				mr.init();
			} catch (Exception e) {
				log.error("Error on init roller of recipient queue {}", rq, e);
				return new ReplyInfo<Object>(e.getMessage());
			}
		}
		for (int i = 0; i < rq.getThreads(); i++) {
			threadPool.execute(rq);
		}
		return ReplyInfo.SUCCESS;
	}

	public MessageService getConsumer(String name) throws NoSuchBeanDefinitionException {
		MessageService service = serviceMap.get(name);
		if (service != null) {
			return service;
		}
		synchronized (serviceMap) {
			service = serviceMap.get(name);
			if (service != null) {
				return service;
			}
			if (appCtx.containsBean(name)) {
				service = appCtx.getBean(name, MessageService.class);
			}
			if (service != null) {
				serviceMap.put(name, service);
			}
		}
		return service;
	}

	public Map<String, TargetMQ> getTargetQueueMap() {
		return targetQueueMap;
	}

	public void setTargetQueueMap(Map<String, TargetMQ> targetQueueMap) {
		this.targetQueueMap = targetQueueMap;
	}

	public Map<String, RecipientMQ> getRecipientQueueMap() {
		return recipientQueueMap;
	}

	public void setRecipientQueueMap(Map<String, RecipientMQ> recipientQueueMap) {
		this.recipientQueueMap = recipientQueueMap;
	}

	public String getCharset() {
		return charset;
	}

	public void setCharset(String charset) {
		this.charset = charset;
	}

}